package com.yhcui.rabbit.rocetmq.demo;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.CountDownLatch2;
import org.apache.rocketmq.common.message.Message;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

/**
 * 异步发送
 */
public class AsyncProducer {

    public static void main(String[] args) throws  Exception {
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer("asynproducer");
        defaultMQProducer.setNamesrvAddr("localhost:9876");
        defaultMQProducer.setSendMsgTimeout(60000);
//        defaultMQProducer.setCreateTopicKey("topic");
        defaultMQProducer.start();
        defaultMQProducer.setRetryTimesWhenSendAsyncFailed(0);
        int msgCount = 100;
        final CountDownLatch2 countDownLatch2 = new CountDownLatch2(msgCount);
        for (int i = 0; i < msgCount; i++) {
            final int index = i;
            Message msg = new Message("topic" , "TAG",  "Order", "Hello".getBytes(StandardCharsets.UTF_8));
            defaultMQProducer.send(msg, new SendCallback() {

                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.printf("%-10d OK %s %n", index,
                            sendResult.getMsgId());
                }

                @Override
                public void onException(Throwable e) {
                    System.out.printf("%-10d Exception %s %n", index, e);
                    e.printStackTrace();
                }
            });

        }

        // 等待5s
        countDownLatch2.await(20, TimeUnit.SECONDS);
        // 如果不再发送消息，关闭Producer实例。
        defaultMQProducer.shutdown();

    }
}
